package spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Top3 {
  //  agent.log文件：时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
  //  1516609143867 province06 city07 user64 add16
  //  需求描述:统计出每一个省份每个广告被点击数量排行的Top3

  def main(args: Array[String]): Unit = {
    //1 初始化spark配置信息并建立与spark的连接
    val conf = new SparkConf().setMaster("local[*]").setAppName(Top3.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    //2 读取数据
    val RDD1: RDD[String] = sc.textFile("dataSet/agent.log")
    //3 拼1：((省份，广告)，1)
    val RDD2: RDD[((String, String), Int)] = RDD1.map {
      x =>
        val fields: Array[String] = x.split(" ")
        ((fields(1), fields(4)), 1)
    }
    //4 聚合,统计每个省份的每个广告点击次数：((省份，广告)，sum)
    val RDD3: RDD[((String, String), Int)] = RDD2.reduceByKey(_ + _)
    //5 将省份作为key，广告加点击数为value：(Province,(AD,sum))
    val RDD4: RDD[(String, (String, Int))] = RDD3.map(x => (x._1._1, (x._1._2, x._2)))
    //6 将同一个省份的数据进行聚合
    val RDD5: RDD[(String, Iterable[(String, Int)])] = RDD4.groupByKey()
    //7 对同一个省份所有广告的集合进行排序并取前3条，排序规则为广告点击总数
    //    方法一
    val RDD6: RDD[(String, List[(String, Int)])] = RDD5.mapValues {
      x => x.toList.sortWith((x, y) => x._2 > y._2).take(3)
    }
    //    方法二
    //    val RDD6: RDD[(String, List[(String, Int)])] = RDD5.map({
    //      group => (group._1, group._2.toList.sortBy(-_._2).take(3))
    //    })
    //8 按照省份ID排序
    val RDD7: RDD[(String, List[(String, Int)])] = RDD6.sortByKey()
    //9 输出结果
    RDD7.collect().foreach(println)
    //10 关闭与spark的连接
    sc.stop()

  }
}
